import os

from coin.base.param_util import to_list
import coin.base.config_util as config_util


class KafkaConfig(object):
  def __init__(self, *, kafka_servers, topic_prefix=None):
    self._kafka_servers = to_list(kafka_servers)
    assert len(self._kafka_servers) > 0
    self._topic_prefix = topic_prefix

  @property
  def kafka_servers(self):
    return self._kafka_servers

  def gen_topic(self, topic):
    if topics is None:
      return None
    assert isinstance(topic, str)
    if self._topic_prefix is not None:
      return "%s%s" % (self._topic_prefix, topic)
    else:
      return topic

  def gen_topics(self, topics):
    if topics is None:
      return None
    if topics is str:
      topics = topics.split(',')
    assert isinstance(topics, list)
    if self._topic_prefix is not None:
      return ["%s%s" % (self._topic_prefix, topic) for topic in topics]
    else:
      return topics

  @staticmethod
  def from_kafka_servers(kafka_servers):
    return KafkaConfig(kafka_servers=kafka_servers.split(','))

  @staticmethod
  def from_config(config_filename):
    config = config_util.load_json(config_filename)
    return KafkaConfig(kafka_servers=config['kafka_servers'].split(','),
                       topic_prefix=config.get('topic_prefix', None))

  @staticmethod
  def from_cmd_config(config_filename):
    if config_filename is None or config_filename == "":
      return KafkaConfig(kafka_servers=['localhost:9092'])
    elif os.path.exists(config_filename):
      return KafkaConfig.from_config(config_filename)
    else:
      return KafkaConfig(kafka_servers=config_filename.split(','))
